Flink水位线不触发问题 |
您所在的位置：网站首页 › flink watermark 多并行度 › Flink水位线不触发问题 |
Flink水位线不触发问题
|
窗口计算时遇到好几次水位线不触发的情况，简单总结下。
首先，介绍下Flink的事件时间(EventTime)和水位线(Watermarks)的概念。 一、处理时间如果要构造一个实时的流式应用，或早或晚都会接触到EventTime这个概念。现实场景中也会遇到消息乱序到达，这里会介绍到为什么需要事件时间和如何去处理乱序到达的数据。 ProcessingTime是Flink系统处理这条消息的时间，EventTime可以理解成是这条消息真实发生的时间。 举个例子，创建一个SlidingWindow，窗口大小为10秒，步长为5秒。关于窗口的更多概念，可以参考Flink官方文档——Windows。 案例1：消息都按时到达 val text = senv.socketTextStream("localhost", 9999) val counts = text.map {(m: String) => (m.split(",")(0), 1) } .keyBy(0) .timeWindow(Time.seconds(10), Time.seconds(5)) .sum(1) counts.print senv.execute("ProcessingTime processing example")如果source中有三条消息，对应的事件时间分别为13秒、13秒和16秒：[站外图片上传中...(image-1e2304-1550219427900)] 它们会落到正确的窗口上，如下图所示。13秒产生的两条消息会落到窗口1[5s-15s]和窗口2[10s-20s]上，16秒产生的消息会落到窗口2[10s-20s]和窗口3[15s-25s]上。最后窗口fire掉时，三个窗口的count值分别为：(a,2), (a,3) and (a,1) ，和预期一致。
pr_ino_windows 案例2：消息delay 其中一条13秒产生的消息晚到了6s，那按上面的代码逻辑，这些消息会落到下面的窗口中：
pr_ooo_windows 延迟的消息会落到窗口2[10s-20s]和窗口3[15s-25s]上。这看起来对窗口2没有影响，因为结果都是3，但窗口3的结果却不一致了。 二、事件时间 因此此处我们采用事件时间，这里水位线的事件时间为当前系统的时间，当然你可以改成数据中的某个时间。 class TimestampExtractor extends AssignerWithPeriodicWatermarks[String] with Serializable { override def extractTimestamp(e: String, prevElementTimestamp: Long) = { e.split(",")(1).toLong } override def getCurrentWatermark(): Watermark = { new Watermark(System.currentTimeMillis) } } senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) val text = senv.socketTextStream("localhost", 9999) .assignTimestampsAndWatermarks(new TimestampExtractor) val counts = text.map {(m: String) => (m.split(",")(0), 1) } .keyBy(0) .timeWindow(Time.seconds(10), Time.seconds(5)) .sum(1) counts.print senv.execute("EventTime processing example")结果如下图所示：[站外图片上传中...(image-b2feb1-1550219427900)] 结果看起来好很多，窗口2和3都正确了，但窗口1却丢了一条数据。 Flink没有将delay的数据分配给窗口3是因为现在是检查消息的事件时间，因此不会放入窗口三中。而没有分配给窗口1的原因是delayed的消息到达系统的时间是19秒，窗口1已经fire掉了。此处就需要watermarks了。 三、水位线（水印）我认为水位线是很重要和有趣的一个概念，我这里会大概描述下，如果想了解更多可以看Google一场精彩的talk，也可以看这个dataArtisans的blog。水位线简单理解就是一个timestamp，当Flink收到这个水印时，Flink理解会收到来自这个时间点之后的消息，也可以理解成告诉Flink运行到哪个事件时间了。 在这个案例中，其实就是告诉Flink一条消息可以迟到多久。 我现在设置水位线为现在的事件提前5秒，相当于告诉Flink我的消息可以迟到五秒。 override def getCurrentWatermark(): Watermark = { new Watermark(System.currentTimeMillis - 5000) }结果变成下图所示: [站外图片上传中...(image-6dfb6e-1550219427900)] 四、允许延迟(Lateness)如果采用“watermark - delay”，如果水位线不超过window_length + delay是不会被fire掉的，所以此刻可以采用allowedLateness方法。在window_end_time + allowed lateness之前，Flink都不会丢弃这条数据。 当消息到达时，Flink会提取它的时间，然后判断它是否在有效的延迟时间内，然后去判断是否fire掉窗口。 但是通过这种途径，一个窗口有可能被fire掉多次，如果需要exactly once processing的话，需要保证sink是幂等的。 五、水位线怎么不触发？数据一直有序得进来，为什么没有窗口被fire掉？没有结果产出？ case1：提取时间失败笔者和上游约定好了数据格式，extractTimestamp中提取的是某个字段为事件时间。研究数据发现约定好的字段突然不发了。 case2：提取时间有了，但是照样失败上游按约定发了该字段后，系统在测试环境运行了一段时间，又没有结果产出了。 调试发现提取时间正常，checkAndGetNextWatermark也正常，但是为什么窗口没被fire掉呢。 因为时间的format变了，由到毫秒的timestamp变成了yyyymmddHHmmss，需要转成timestamp。 case3：一切正常，窗口不fire EventTimeTrigger.java @Override public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception { if (window.maxTimestamp() = lastEmittedWatermark) { lastEmittedWatermark = potentialWM; } return new Watermark(lastEmittedWatermark); }debug发现lastEmittedWatermark确实有更新，这说明这个地方是触发了Watermark的值。但是debug的过程中，发现时不时会出现初始值的水位线。 SystemProcessingTimeService.java TimestampsAndPeriodicWatermarksOperator.java @Override public void onProcessingTime(long timestamp) throws Exception { // register next timer Watermark newWatermark = userFunction.getCurrentWatermark(); if (newWatermark != null && newWatermark.getTimestamp() > currentWatermark) { currentWatermark = newWatermark.getTimestamp(); // emit watermark output.emitWatermark(newWatermark); } long now = getProcessingTimeService().getCurrentProcessingTime(); getProcessingTimeService().registerTimer(now + watermarkInterval, this); }TimestampsAndPeriodicWatermarksOperator会做判断：如果新的水位线小于当前的水位线，就不会更新了。 终于，顺着StreamInputProcessor–>StatusWatermarkValve理了下来，看见这样的处理逻辑: StreamInputProcessor.java StatusWatermarkValve.java private void findAndOutputNewMinWatermarkAcrossAlignedChannels() { long newMinWatermark = Long.MAX_VALUE; // determine new overall watermark by considering only watermark-aligned channels across all channels for (InputChannelStatus channelStatus : channelStatuses) { if (channelStatus.isWatermarkAligned) { newMinWatermark = Math.min(channelStatus.watermark, newMinWatermark); } } // we acknowledge and output the new overall watermark if it is larger than the last output watermark if (newMinWatermark > lastOutputWatermark) { lastOutputWatermark = newMinWatermark; outputHandler.handleWatermark(new Watermark(lastOutputWatermark)); } }这里会将所有的channel status的水位线做个汇总：取最小的水位线。那是不是问题出在这里？后面debug了下看看，确实，这个地方有的channel status下的水位线一直是最小的long值那个不正常的水位线，进而导致整体的水位线发送不出去。 那么为什么会出现这种情况呢，百思不得其解。 当 [window_start_time,window_end_time) 有数据，watermark Time大于等于window_end_time时，会触发window trigger。 因为之前运行都是正常的，检查了数据也没问题。去翻改动，有影响的可能就是改了一些算子的并行度。 assigntimestampandwatermarks和map的并行度一样了就不能生成水位线了？ 于是修改了assigntimestampandwatermarks的并行度，window正常fire掉了。 分析原因： Flink source用到了FlinkKafkaConsumer010，没有指定KafkaPartitioner的话，会通过FixedPartitioner来给出默认的partitioner方法： public int partition(T record, byte[] key, byte[] value, String targetTopic, int[] partitions) { Preconditions.checkArgument(partitions != null && partitions.length > 0, "Partitions of the target topic is empty."); return partitions[this.parallelInstanceId % partitions.length]; }parallelInstanceId代表着Flink consumer程序的并行度ID，假如FlinkKafkaConsumer010的并行度是12，那么这12个线程的ID分别是1-12. parallelInstances代表着总的并行度，即12。 partitions是一个kafka partition的数组，kafka的topic的partitions是4（因为性能测试，换到只有一个节点的kafka）。 Flink partition的规则，就是Flink的并行度ID除以kafka partition length取余。 因此kafka编号为1-4的partition分别对应source node的1-4的partition，那么source node5-12的partition就为空了。 默认的partition策略是按照Flink的并行度ID与kafka中partition的数量取余的方法分配的，而与数据本身没有关系。 source node的partition为5-12的接收不到数据， Watermark的生成是数据驱动的，只有当TimestampAndWatermarkAssigner”看到”数据时，watermark才会生效。 而map和assigntimestampandwatermarks并发度一样的话，这八个partition的watermark不会修改，所以会出现watermark的初始值一直存在的情况。 当assigntimestampandwatermarks的并行度修改后，事件会被shuffled across（洗牌），因此到了TimestampAndWatermarkAssigner不会有空的partition存在了。 以上。 五 实际定位 问题：发现程序没有输出，没有任何报错。可以先去任务的Appmaster上看watermark运行情况.来定位是否是watermarke的问题
如果是显示no watermark ，按照上述的case 进行诊断。在assignTimestampsAndWatermarks 后setParallelism(2) 。具体并行度根据实际需求设置，在这里kafkatopic只有2个分区，并行度设置为2
|
【本文地址】
今日新闻 |
推荐新闻 |